package com.bff.gaia.mix.api.xjoin2.client.rdb.async;

import com.alibaba.fastjson.JSONObject;
import io.vertx.core.json.JsonArray;
import io.vertx.ext.sql.SQLClient;
import io.vertx.ext.sql.SQLConnection;
import com.bff.gaia.mix.api.functions.async.ResultFuture;
import com.bff.gaia.mix.api.xjoin2.core.cache.AsyncCacheObject;
import com.bff.gaia.mix.api.xjoin2.core.enums.AsyncXjoinCacheContentType;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.functions.RichAsyncXjoinFunction;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

import static com.bff.gaia.mix.api.xjoin2.core.utils.KeyUtil.buildKey;

public class RdbRichAsyncXjoinFunction extends RichAsyncXjoinFunction {

	private static final Logger LOG = LoggerFactory.getLogger(RdbRichAsyncXjoinFunction.class);

	public final static int DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE = 1;

	public final static int DEFAULT_VERTX_WORKER_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;

	public final static int DEFAULT_MAX_DB_CONN_POOL_SIZE = DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE + DEFAULT_VERTX_WORKER_POOL_SIZE;

	private transient SQLClient rdbSQLClient;

	public RdbRichAsyncXjoinFunction(StreamTableInfo streamTableInfo, SideTableInfo sideTableInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList) {
		super(new RdbAsyncXjoinInfo(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList));
	}

	@Override
	public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {

		JsonArray inputParams = new JsonArray();
		for (Integer conValIndex : xjoinInfo.getStreamTableKeyFieldPos()) {
			Object equalObj = input.getField(conValIndex);
			if (equalObj == null) {
				resultFuture.complete(null);
				return;
			}
			inputParams.add(equalObj);
		}

		String key = buildKey(input, xjoinInfo.getStreamTableKeyFieldPos());

		if (isCacheOpen()) {

			AsyncCacheObject cacheObject = get(key);
			if (cacheObject != null) {
				if (AsyncXjoinCacheContentType.MissVal == cacheObject.getContentType()) {
					dealMissKey(input, resultFuture);
					return;
				} else if (AsyncXjoinCacheContentType.MultiLine == cacheObject.getContentType()) {
					List<Row> rowList = Lists.newArrayList();
					for (JSONObject jsonArray : cacheObject.getContent()) {
						Row row = fillData(input, jsonArray);
						rowList.add(row);
					}
					resultFuture.complete(rowList);
					return;
				} else {
					throw new RuntimeException("not support cache obj type " + cacheObject.getContentType());
				}
			}
		}
		rdbSQLClient.getConnection(conn -> {
			if (conn.failed()) {
				//Treatment failures
				resultFuture.completeExceptionally(conn.cause());
				return;
			}

			final SQLConnection connection = conn.result();
			String sqlCondition = xjoinInfo.getSqlQuery();
			connection.queryWithParams(sqlCondition, inputParams, rs -> {
				if (rs.failed()) {
					LOG.error("Cannot retrieve the data from the database", rs.cause());
					resultFuture.complete(null);
					return;
				}

				List<JSONObject> cacheContent = Lists.newArrayList();
				int resultSize = rs.result().getResults().size();
				System.out.println("======================================resultSize "+resultSize);
				if (resultSize > 0) {
					List<Row> rowList = Lists.newArrayList();

					for (JsonArray line : rs.result().getResults()) {
						JSONObject jsonObject = new JSONObject();
						for (Map.Entry<Integer, String> entry : xjoinInfo.getSideTableInFieldIndex().entrySet()) {
							//TODO choose the right type rather than all string
							jsonObject.put(entry.getValue(), String.valueOf(line.getValue(entry.getKey())));
						}
						Row row = fillData(input, jsonObject);
						if (isCacheOpen()) {
							cacheContent.add(jsonObject);
						}
						rowList.add(row);
					}

					if (isCacheOpen()) {
						//TODO add a switch to cache miss object
						put(key, AsyncCacheObject.createAsyncCacheObject(AsyncXjoinCacheContentType.MultiLine, cacheContent));
					}

					resultFuture.complete(rowList);
				} else {
					dealMissKey(input, resultFuture);
					if (isCacheOpen()) {
						put(key, AsyncCacheObject.createAsyncCacheObject(AsyncXjoinCacheContentType.MissVal));
					}
				}

				// and close the connection
				connection.close(done -> {
					if (done.failed()) {
						throw new RuntimeException(done.cause());
					}
				});
			});
		});

	}

	@Override
	public Row fillData(Row stream, JSONObject side) {
		Row output = new Row(xjoinInfo.getOutFieldSize());
		for (Map.Entry<Integer, Integer> entry : xjoinInfo.getStreamTableOutFieldIndex().entrySet()) {
			output.setField(entry.getKey(), stream.getField(entry.getValue()));
		}
		for (Map.Entry<Integer, String> entry : xjoinInfo.getSideTableOutFieldIndex().entrySet()) {
			if (side == null) {
				output.setField(entry.getKey(), null);
			} else {
				output.setField(entry.getKey(), side.get(entry.getValue()));
			}
		}
		return output;
	}

	@Override
	public void close() throws Exception {
		super.close();
		if (rdbSQLClient != null) {
			rdbSQLClient.close();
		}

	}

	public void setRdbSQLClient(SQLClient rdbSQLClient) {
		this.rdbSQLClient = rdbSQLClient;
	}
}